package pers.amos.eshop.advance.data.sync.mq;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.log.StaticLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import pers.amos.eshop.advance.data.sync.common.DataTypeConstants;
import pers.amos.eshop.advance.data.sync.common.EventTypeConstants;
import pers.amos.eshop.advance.data.sync.redis.RedisUtils;
import pers.amos.eshop.advance.data.sync.service.ProductService;

import java.util.ArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author amos wong
 * @create 2020-12-17 11:13 上午
 *
 * 模板方法模式
 */
@Component
public abstract class AbstractQueueReceiver {

    @Autowired
    private ProductService productService;

    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private RabbitMqSender rabbitMqSender;

    /**
     * 用于去重
     */
    private final CopyOnWriteArraySet<String> removeDuplicationSet = new CopyOnWriteArraySet<>();

    private static final ArrayList<Long> idList = CollUtil.newArrayList();

    private static final String BRAND_PREFIX_KEY = "brand_";
    private static final String CATEGORY_PREFIX_KEY = "category_";
    private static final String PRODUCT_INTRO_PREFIX_KEY = "product_intro_";
    private static final String PRODUCT_PROPERTY_PREFIX_KEY = "brand_property_";
    private static final String PRODUCT_PREFIX_KEY = "product_";
    private static final String PRODUCT_SPECIFICATION_PREFIX_KEY = "product_specification_";

    /**
     * 构造方法中开启定时任务 发送去重队列中的消息
     */
    public AbstractQueueReceiver() {
        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
            if (CollUtil.isNotEmpty(removeDuplicationSet)) {
                for (String message : removeDuplicationSet) {
                    sendMessage(message);
                    StaticLog.info("【数据同步服务】向聚合服务发送一条经过去重的消息, message = {}",
                            message);
                }
            }

            removeDuplicationSet.clear();
        }, 0, 1000, TimeUnit.MILLISECONDS);
    }

    public void process(String message) {
        StaticLog.info("【数据同步服务】MQ 收到数据变更消息:{}", message);

        JSONObject jsonObject = new JSONObject(message);
        String eventType = (String) jsonObject.get("event_type");
        String dataType = (String) jsonObject.get("data_type");
        Long id = Convert.toLong(jsonObject.get("id"));
        Long productId = Convert.toLong(jsonObject.get("product_id"));

        if (DataTypeConstants.BRAND_DATA_TYPE.equals(dataType)) {
            processBrandDataChange(id, eventType);

            String sendMessage = rabbitMqSender.buildSendMessage("brand", id);
            sendMessage(sendMessage);
        } else if (DataTypeConstants.CATEGORY_DATA_TYPE.equals(dataType)) {
            processCategoryDataChange(id, eventType);

            String sendMessage = rabbitMqSender.buildSendMessage("category", id);
            sendMessage(sendMessage);
        } else if (DataTypeConstants.PRODUCT_INTRO_DATA_TYPE.equals(dataType)) {
            processProductIntroDataChange(id, eventType);

            String sendMessage = rabbitMqSender.buildSendMessage("product_intro", productId);
            sendMessage(sendMessage);
        } else if (DataTypeConstants.PRODUCT_PROPERTY_DATA_TYPE.equals(dataType)) {
            processProductPropertyDataChange(id, eventType);

            String sendMessage = rabbitMqSender.buildSendMessage("product", productId);
            removeDuplicationSet.add(sendMessage);
            StaticLog.info("【商品服务-商品属性维度】向去重内存队列中放入了一条数据, productId = {}",
                    id);
        } else if (DataTypeConstants.PRODUCT_DATA_TYPE.equals(dataType)) {
            processProductDataChange(id, eventType);

            String sendMessage = rabbitMqSender.buildSendMessage("product", id);
            removeDuplicationSet.add(sendMessage);
            StaticLog.info("【商品服务-商品主属性维度】向去重内存队列中放入了一条数据, productId = {}",
                    id);
        } else {
            processProductSpecificationDataChange(id, eventType);
            String sendMessage = rabbitMqSender.buildSendMessage("product", productId);
            removeDuplicationSet.add(sendMessage);
            StaticLog.info("【商品服务-商品规格参数维度】向去重内存队列中放入了一条数据, productId = {}",
                    id);
        }
    }

    /**
     * 发送消息的抽象方法，由子类去实现
     *
     * @param message 具体发送的消息
     */
    protected abstract void sendMessage(String message);

    private void processBrandDataChange(Long id, String eventType) {
        if (eventType.equals(EventTypeConstants.ADD_EVENT_TYPE) || eventType.equals(EventTypeConstants.UPDATE_EVENT_TYPE)) {
            String brand = productService.findBrandById(id);
            redisUtils.set(BRAND_PREFIX_KEY + id, brand);
        } else {
            redisUtils.delete(BRAND_PREFIX_KEY + id);
        }
    }

    private void processCategoryDataChange(Long id, String eventType) {
        if (eventType.equals(EventTypeConstants.ADD_EVENT_TYPE) || eventType.equals(EventTypeConstants.UPDATE_EVENT_TYPE)) {
            String category = productService.findCategoryById(id);
            redisUtils.set(CATEGORY_PREFIX_KEY + id, category);
        } else {
            redisUtils.delete(CATEGORY_PREFIX_KEY + id);
        }
    }

    private void processProductIntroDataChange(Long id, String eventType) {
        if (eventType.equals(EventTypeConstants.ADD_EVENT_TYPE) || eventType.equals(EventTypeConstants.UPDATE_EVENT_TYPE)) {
            String productIntro = productService.findProductIntroById(id);
            // TODO 商品描述字段比较大，这里需要采用分段存储
            redisUtils.set(PRODUCT_INTRO_PREFIX_KEY + id, productIntro);
        } else {
            redisUtils.delete(PRODUCT_INTRO_PREFIX_KEY + id);
        }
    }

    private void processProductPropertyDataChange(Long id, String eventType) {
        if (eventType.equals(EventTypeConstants.ADD_EVENT_TYPE) || eventType.equals(EventTypeConstants.UPDATE_EVENT_TYPE)) {
            String productProperty = productService.findProductPropertyById(id);
            redisUtils.set(PRODUCT_PROPERTY_PREFIX_KEY + id, productProperty);
        } else {
            redisUtils.delete(PRODUCT_PROPERTY_PREFIX_KEY + id);
        }
    }

    private void processProductDataChange(Long id, String eventType) {
        if (eventType.equals(EventTypeConstants.ADD_EVENT_TYPE) || eventType.equals(EventTypeConstants.UPDATE_EVENT_TYPE)) {
            String product = productService.findProductById(id);
            redisUtils.set(PRODUCT_PREFIX_KEY + id, product);
        } else {
            redisUtils.delete(PRODUCT_PREFIX_KEY + id);
        }
    }

    private void processProductSpecificationDataChange(Long id, String eventType) {
        if (eventType.equals(EventTypeConstants.ADD_EVENT_TYPE) || eventType.equals(EventTypeConstants.UPDATE_EVENT_TYPE)) {
            String productSpecification = productService.findProductSpecificationById(id);
            redisUtils.set(PRODUCT_SPECIFICATION_PREFIX_KEY + id, productSpecification);
        } else {
            redisUtils.delete(PRODUCT_SPECIFICATION_PREFIX_KEY + id);
        }
    }

    /**
     * 远程调用接口查询品牌信息 这里使用批量查询来优化性能
     * 批量查询看看就好 还有很多细节问题需要优化
     *
     * @param id
     */
    private void batchSyncBrandData(Long id) {
        if (idList.size() < 2) {
            idList.add(id);
            StaticLog.info("【数据同步服务-同步品牌信息接口】向 idList 中添加数据 id = {}", id);
            return;
        }

        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < idList.size(); i++) {
            if (i == idList.size() - 1) {
                sb.append(idList.get(i));
                break;
            }
            sb.append(idList.get(i));
            sb.append(",");
        }

        String brandList = productService.findBrandListByIds(sb.toString());

        JSONArray jsonArray = new JSONArray(brandList);

        for (int i = 0; i < jsonArray.size(); i++) {
            JSONObject jsonObject = jsonArray.getJSONObject(i);
            StaticLog.info("【数据同步服务-同步品牌信息接口】向 redis 中批量新增数据 id = {}", jsonObject.get("id"));
            redisUtils.set(BRAND_PREFIX_KEY + id, jsonObject.toString());
        }

        idList.clear();
    }

}
